package com.electric.flinkcep.service;

import com.electric.flinkcep.model.Event;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.util.concurrent.TimeUnit;

/**
 * CEP模式定义服务
 * 负责定义各种复杂事件处理模式
 */
public class CEPPatternService implements Serializable {
    
    private static final Logger logger = LoggerFactory.getLogger(CEPPatternService.class);
    
    /**
     * 创建连续数字1检测模式
     * 检测连续3次出现数字1的事件
     */
    public static Pattern<Event, ?> createConsecutiveOnesPattern() {
        return Pattern.<Event>begin("first")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        return "1".equals(event.getData());
                    }
                })
                .followedBy("second")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        return "1".equals(event.getData());
                    }
                })
                .followedBy("third")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        return "1".equals(event.getData());
                    }
                })
                .within(Time.seconds(30));
    }

    /**
     * 创建高频告警模式
     * 检测在30秒内同一用户产生3个或以上的ERROR级别事件
     */
    public static Pattern<Event, ?> createHighFrequencyErrorPattern() {
        return Pattern.<Event>begin("first")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        return "ERROR".equals(event.getSeverity());
                    }
                })
                .next("second")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        return "ERROR".equals(event.getSeverity());
                    }
                })
                .next("third")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        return "ERROR".equals(event.getSeverity());
                    }
                })
                .within(Time.seconds(30));
    }
    
    /**
     * 创建异常登录模式
     * 检测在5分钟内同一用户有超过5次登录失败事件
     */
    public static Pattern<Event, ?> createAbnormalLoginPattern() {
        return Pattern.<Event>begin("loginFailures")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        return "LOGIN_FAILURE".equals(event.getType());
                    }
                })
                .timesOrMore(5)
                .within(Time.minutes(5));
    }
    
    /**
     * 创建系统异常模式
     * 检测在1分钟内出现CRITICAL级别事件后紧跟着多个ERROR事件
     */
    public static Pattern<Event, ?> createSystemAnomalyPattern() {
        return Pattern.<Event>begin("critical")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        return "CRITICAL".equals(event.getSeverity());
                    }
                })
                .followedBy("errors")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        return "ERROR".equals(event.getSeverity());
                    }
                })
                .timesOrMore(2)
                .within(Time.minutes(1));
    }
    
    /**
     * 创建数据异常模式
     * 检测在10秒内同一数据源产生的异常数据事件
     */
    public static Pattern<Event, ?> createDataAnomalyPattern() {
        return Pattern.<Event>begin("dataAnomaly")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        return "DATA_ANOMALY".equals(event.getType()) && 
                               "HIGH".equals(event.getSeverity());
                    }
                })
                .timesOrMore(3)
                .within(Time.seconds(10));
    }
    
    /**
     * 创建性能降级模式
     * 检测在2分钟内出现性能警告后跟随系统错误的模式
     */
    public static Pattern<Event, ?> createPerformanceDegradationPattern() {
        return Pattern.<Event>begin("performanceWarning")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        return "PERFORMANCE_WARNING".equals(event.getType());
                    }
                })
                .followedBy("systemError")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        return "SYSTEM_ERROR".equals(event.getType()) && 
                               "ERROR".equals(event.getSeverity());
                    }
                })
                .within(Time.minutes(2));
    }
    
    /**
     * 创建安全威胁模式
     * 检测在30秒内出现多次安全相关事件
     */
    public static Pattern<Event, ?> createSecurityThreatPattern() {
        return Pattern.<Event>begin("securityEvents")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        return event.getType() != null && 
                               (event.getType().contains("SECURITY") || 
                                event.getType().contains("UNAUTHORIZED") ||
                                event.getType().contains("INTRUSION"));
                    }
                })
                .timesOrMore(2)
                .within(Time.seconds(30));
    }
    
    /**
     * 创建通用告警模式
     * 可配置的通用模式，用于检测指定类型和严重级别的事件
     */
    public static Pattern<Event, ?> createGenericAlertPattern(String eventType, String severity, int count, long timeWindowSeconds) {
        logger.info("创建通用告警模式: 事件类型={}, 严重级别={}, 次数={}, 时间窗口={}秒", 
                   eventType, severity, count, timeWindowSeconds);
        
        return Pattern.<Event>begin("genericAlert")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        boolean typeMatch = eventType == null || eventType.equals(event.getType());
                        boolean severityMatch = severity == null || severity.equals(event.getSeverity());
                        return typeMatch && severityMatch;
                    }
                })
                .timesOrMore(count)
                .within(Time.of(timeWindowSeconds, TimeUnit.SECONDS));
    }
    
    /**
     * 创建复合条件模式
     * 检测满足多个条件的复杂事件模式
     */
    public static Pattern<Event, ?> createComplexConditionPattern() {
        return Pattern.<Event>begin("start")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        // 检测高优先级事件或来自特定源的事件
                        return "HIGH".equals(event.getSeverity()) || 
                               "CRITICAL".equals(event.getSeverity()) ||
                               (event.getSource() != null && event.getSource().startsWith("PROD"));
                    }
                })
                .followedBy("related")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event event) {
                        // 检测相关的错误事件
                        return "ERROR".equals(event.getSeverity()) && 
                               event.getData() != null && 
                               event.getData().contains("exception");
                    }
                })
                .within(Time.minutes(5));
    }
}